package com.dabai.iot.mqtt.dubbo;

import cn.hutool.core.date.DateUtil;
import com.dabai.common.core.utils.Func;
import com.dabai.common.core.utils.balance.QueueBalance;
import com.dabai.iot.mqtt.api.RemoteMqttService;
import com.dabai.iot.mqtt.model.DeviceConnectionSession;
import com.dabai.iot.mqtt.model.MqttMessageContext;
import com.dabai.iot.mqtt.model.ServerClient;
import com.dabai.iot.mqtt.model.ServerConnectionSession;
import com.dabai.iot.mqtt.server.processor.MessageToDeviceProcessor;
import com.dabai.iot.mqtt.server.processor.MessageToServerProcessor;
import com.dabai.iot.mqtt.util.ClientUtil;
import com.dabai.iot.mqtt.util.MqttSessionManager;
import com.dabai.iot.mqtt.util.ServerConnectionSessionManager;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * MQTT 服务实现类
 *
 * @author kev1n
 */
@Slf4j
@Service
@DubboService
@RequiredArgsConstructor
public class RemoteMqttServiceImpl implements RemoteMqttService {

    private final MessageToDeviceProcessor messageToDeviceProcessor;
    private final MessageToServerProcessor messageToServerProcessor;

    @Override
    public DeviceConnectionSession getDeviceConnectionSession(String productKey, String deviceSn) {
        String clientId = ClientUtil.deviceClientId(productKey, deviceSn);
        return MqttSessionManager.getInstance().getSession(clientId);
    }

    @Override
    public void closeDeviceConnectionSession(String traceId, String productKey, String deviceSn) {
        String clientId = ClientUtil.deviceClientId(productKey, deviceSn);
        DeviceConnectionSession session = MqttSessionManager.getInstance().getSession(clientId);
        if (session != null) {
            // 移除连接会话
            MqttSessionManager.getInstance().removeSession(clientId);
            // 关闭客户端连接
            log.info("IOT Server ====> 设备已删除，断开连接，clientId: " + clientId);
            session.getMqttEndpoint().close();
        }
    }

    @Override
    public List<ServerClient> getServerClientList(String groupID) {
        List<ServerConnectionSession> sessionList = ServerConnectionSessionManager.getInstance().getSession(groupID);
        if (Func.isEmpty(sessionList)) {
            return List.of();
        }

        List<ServerClient> serverClients = new ArrayList<>();
        for (ServerConnectionSession session : sessionList) {
            ServerClient serverClient = ClientUtil.parseClientId(session.getMqttEndpoint().clientIdentifier());
            serverClient.setAddress(session.getMqttEndpoint().remoteAddress().toString());
            serverClient.setOnlineTime(DateUtil.formatDateTime(session.getOnlineDate()));
            serverClients.add(serverClient);
        }

        return serverClients;
    }

    @Override
    public void sendMessageToDevice(String traceId, String productKey, String deviceSn, String topic, String payload) {
        String clientId = ClientUtil.deviceClientId(productKey, deviceSn);
        DeviceConnectionSession session = MqttSessionManager.getInstance().getSession(clientId);
        if (session == null) {
            return;
        }

        MqttMessageContext mqttMessageContext = new MqttMessageContext();
        mqttMessageContext.setMqttEndpoint(session.getMqttEndpoint());
        mqttMessageContext.setTraceId(traceId);
        mqttMessageContext.setTopic(topic);
        mqttMessageContext.setPayload(payload);
        mqttMessageContext.setQosLevel(MqttQoS.AT_MOST_ONCE);

        messageToDeviceProcessor.add(mqttMessageContext);
    }

    @Override
    public void sendMessageToServer(String traceId, Long consumerId, String groupID, String topic, String payload) {
        // 1、查询是否有客户端(业务平台)使用该消费组接入（一对多）
        List<ServerConnectionSession> sessionList = ServerConnectionSessionManager.getInstance().getSession(groupID);

        if (Func.isNotEmpty(sessionList)) { // 有连接的订阅服务端

            // 2、向订阅的客户端发布消息（顺序轮询模式）
            QueueBalance<ServerConnectionSession> queueBalance = new QueueBalance<>();
            ServerConnectionSession session = queueBalance.chooseOne(sessionList);

            // 3、发送消息到业务端
            MqttMessageContext mqttMessageContext = new MqttMessageContext();
            mqttMessageContext.setMqttEndpoint(session.getMqttEndpoint());
            mqttMessageContext.setTraceId(traceId);
            mqttMessageContext.setTopic(topic);
            mqttMessageContext.setPayload(payload);
            mqttMessageContext.setQosLevel(MqttQoS.AT_MOST_ONCE);

            messageToServerProcessor.add(mqttMessageContext);
            log.info("服务端订阅 ======> 消费组【{}】有服务端订阅，向服务端推送消息", consumerId);
        } else {
            log.info("服务端订阅 ======> 消费组【{}】无服务端订阅，不需要处理", consumerId);
        }
    }
}
